package top.wangjiu.consume.spring;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import reactor.core.publisher.Flux;
import top.wangjiu.reactive.rpc.registration.LocalConfigServerAddrList;
import top.wangjiu.reactive.rpc.registration.ServerAddrList;
import top.wangjiu.reactive.rpc.spring.client.RpcClient;
import top.wangjiu.service.interfaces.OneService;
import top.wangjiu.service.interfaces.TwoService;

import java.util.concurrent.atomic.AtomicInteger;

@SpringBootApplication
public class ConsumeApplication {

	private static final Log log = LogFactory.getLog(ConsumeApplication.class);

	@RpcClient
	private OneService oneService;

	@RpcClient
	private OneService oneService2;

	@RpcClient(version = 1)
	private OneService oneService3;

	@RpcClient
	private TwoService twoService;

	@Autowired
	private TwoService twoService2;


	public static void main(String[] args) {

		ConfigurableApplicationContext run = SpringApplication.run(ConsumeApplication.class, args);
		ConsumeApplication application = run.getBean(ConsumeApplication.class);
		application.test();
	}

	@Bean
	public ServerAddrList serverAddrList(){
		LocalConfigServerAddrList addrList = new LocalConfigServerAddrList();
		addrList.addService(OneService.class,0,"localhost",9000);
		addrList.addService(TwoService.class,0,"localhost",9000);
		return addrList;
	}


	void test(){
		//响应式，Flux,测试回压功能，
		Thread[] threads = new Thread[5];
		for (int i=0;i<5;i++){
			threads[i] = new Thread(()->{
				int start = 0;
				int end = 10;
				//模拟每次请求3个资源
				int requestPageSize = 3;
				AtomicInteger count = new AtomicInteger(0);
				//向生产者获取从start到end的数字列表，测试回压
				Flux<Integer> list = oneService.getList(start, end);
				list.subscribe(new Subscriber<Integer>() {
					Subscription s = null;
					@Override
					public void onSubscribe(Subscription s) {
						this.s = s;
						//与生产者建立了通道，开启请求第一批数据
						s.request(requestPageSize);
						System.out.println("onSubscribe:"+Thread.currentThread().getId());
					}

					@Override
					public void onNext(Integer num) {
						System.out.println("onNext:"+Thread.currentThread().getId()+",res:"+num);
						try {
							Thread.sleep(500);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
						//模拟处理完一批资源之后，请求下一批资源
						if(num<end && count.incrementAndGet() == requestPageSize){
							s.request(requestPageSize);
							count.set(0);
						}
					}

					@Override
					public void onError(Throwable t) {}

					@Override
					public void onComplete() {
						System.out.println("消费者已经消费完了所有的数据:"+Thread.currentThread().getId());
					}
				});
			});
		}
		for (int i=0;i<5;i++){
			threads[i].start();
		}

		//响应式，Mono
		oneService.sayByeBye("张三").subscribe(System.out::println,null,()-> System.out.println("完成"));
		oneService.sayByeBye("李四").subscribe(System.out::println,Throwable::printStackTrace,()-> System.out.println("完成"));
		oneService.sayByeBye("王五").subscribe(r->{
			if(r==null){
				System.out.println("空");
			}
		},null,()-> System.out.println("完成"));

		//同步形式
		System.out.println(oneService.sayHello());
		System.out.println(twoService.sayByeBye());
		System.out.println(twoService.getSum(1l,2l));

		//future异步形式
		oneService.getSum(3,3).whenComplete((sum,e)->{
			if(e!=null){
				e.printStackTrace();
			}else {
				System.out.println("oneService.getSum："+sum);
			}
		});
	}


}
